package com.jd.platform.gobrs.client.core.key.manual;

import com.alibaba.fastjson.JSON;
import com.jd.platform.gobrs.client.Context;
import com.jd.platform.gobrs.client.core.key.IKeyPusher;
import com.jd.platform.gobrs.client.core.worker.WorkerInfoHolder;
import com.jd.platform.gobrs.client.log.JdLogger;
import com.jd.platform.gobrs.common.model.GCacheKeyModel;
import com.jd.platform.gobrs.common.model.GCacheKeyMsg;
import com.jd.platform.gobrs.common.model.KeyCountModel;
import com.jd.platform.gobrs.common.model.typeenum.MessageType;
import io.netty.channel.Channel;

import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 将msg推送到netty的pusher
 *
 * @author sizegang wrote on 2020-01-06
 * @version 1.0
 */
public class ManualNettyKeyPusher implements IKeyPusher {

    /**
     * 推送热点key
     *
     * @param appName
     * @param list
     */
    @Override
    public void send(String appName, List<GCacheKeyModel> list) {
        //积攒了半秒的key集合，按照hash分发到不同的worker
        long now = System.currentTimeMillis();

        Map<Channel, List<GCacheKeyModel>> map = new HashMap<>();
        for (GCacheKeyModel model : list) {
            model.setCreateTime(now);
            // 选择worker 通道进行上报 worker 是通过ectd 监听到的 然后设置到map中
            Channel channel = WorkerInfoHolder.chooseChannel(model.getKey());
            if (channel == null) {
                continue;
            }

            List<GCacheKeyModel> newList = map.computeIfAbsent(channel, k -> new ArrayList<>());
            newList.add(model);
        }
        if (map.size() == 0) {
            JdLogger.error(getClass(), "--------- No Worker key canot push!!!!  ---------");
        }

        for (Channel channel : map.keySet()) {
            try {
                List<GCacheKeyModel> batch = map.get(channel);
                GCacheKeyMsg hotKeyMsg = new GCacheKeyMsg(MessageType.REQUEST_NEW_KEY, Context.APP_NAME);
                hotKeyMsg.setHotKeyModels(batch);
                // 将热key推到server端
                channel.writeAndFlush(hotKeyMsg).sync();
            } catch (Exception e) {
                try {
                    InetSocketAddress insocket = (InetSocketAddress) channel.remoteAddress();
                    JdLogger.error(getClass(), "flush error " + insocket.getAddress().getHostAddress());
                } catch (Exception ex) {
                    JdLogger.error(getClass(), "flush error");
                }

            }
        }

    }

    /**
     * 推送 key 当前jvm中被命中的次数 push到 worker 之后， worker 会通过滑动窗口算法进行计算 热key数量
     *
     * @param appName
     * @param list
     */
    @Override
    public void sendCount(String appName, List<KeyCountModel> list) {
        //积攒了10秒的数量，按照hash分发到不同的worker
        long now = System.currentTimeMillis();
        Map<Channel, List<KeyCountModel>> map = new HashMap<>();
        for (KeyCountModel model : list) {
            model.setCreateTime(now);
            Channel channel = WorkerInfoHolder.chooseChannel(model.getRuleKey());
            if (channel == null) {
                continue;
            }

            List<KeyCountModel> newList = map.computeIfAbsent(channel, k -> new ArrayList<>());
            newList.add(model);
        }

        for (Channel channel : map.keySet()) {
            try {
                List<KeyCountModel> batch = map.get(channel);
                GCacheKeyMsg hotKeyMsg = new GCacheKeyMsg(MessageType.REQUEST_HIT_COUNT, Context.APP_NAME);
                hotKeyMsg.setKeyCountModels(batch);
                channel.writeAndFlush(hotKeyMsg).sync();
            } catch (Exception e) {
                try {
                    InetSocketAddress insocket = (InetSocketAddress) channel.remoteAddress();
                    JdLogger.error(getClass(), "flush error " + insocket.getAddress().getHostAddress());
                } catch (Exception ex) {
                    JdLogger.error(getClass(), "flush error");
                }

            }
        }
    }

}
